using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Me.Prettyprint.Cassandra.Service;
using Me.Prettyprint.Cassandra.Serializers;
using Me.Prettyprint.Cassandra.Utils;
using Me.Prettyprint.Hector.Api;
using Me.Prettyprint.Hector.Api.Beans;
using Me.Prettyprint.Hector.Api.Ddl;
using Me.Prettyprint.Hector.Api.Factory;
using Me.Prettyprint.Hector.Api.Locking;
using Me.Prettyprint.Hector.Api.Mutation;
using Me.Prettyprint.Hector.Api.Query;

namespace Me.Prettyprint.Cassandra.Locking
{
	public class HLockManagerImpl<K> : AbstractLockManager<K>
	{
        //private ScheduledExecutorService scheduler;
        private long lockTtl = 5000;
        private int colTtl = 5;
        private int maxSelectSize = 10;

        public HLockManagerImpl(Cluster cluster, HLockManagerConfigurator hlc)
            : base(cluster, hlc)
        {
            lockTtl = hlc.LocksTTLInMillis;
            colTtl = (int)(lockTtl / 1000);
            maxSelectSize = hlc.MaxSelectSize;
        }
        public override HLock CreateLock(string lockPath)
        {
            return new HLockImpl(lockPath, GenerateLockId());
        }
        private string GenerateLockId()
        {
            return Guid.NewGuid().ToString();
        }
        private void MaybeSetInternalLockId(HLock hlock)
        {
            if (hlock.LockId == null)
            {
                hlock.LockId = GenerateLockId();
            }
        }
        private Dictionary<string, string> ReadExistingLocks(HLock hlock)
        {
            SliceQuery<String, String, String> sliceQuery = HFactory.CreateSliceQuery(
                Keyspace, StringSerializer.Instance, StringSerializer.Instance, StringSerializer.Instance)
                .SetColumnFamily(LockManagerConfigurator.LockManagerCF).SetKey(hlock.Path);
            
            sliceQuery.SetRange(null, null, false, maxSelectSize);
            QueryResult<ColumnSlice<String, String>> queryResult = sliceQuery.Execute();
            return GetResults(queryResult);
        }
        private Dictionary<string, string> GetResults(QueryResult<ColumnSlice<string, string>> queryResult)
        {
            Dictionary<string, string> result = new Dictionary<string,string>();
            foreach (HColumn<string, string> col in queryResult.Get().GetColumns())
            {
                result[col.Name] = col.Value;
            }
            return result;
        }
        protected void WriteLock(HLock hlock)
        {
            WriteLock(hlock, hlock.LockId.ToString());
        }
        public void WriteLock(HLock hlock, List<string> keySet)
        {
            StringBuilder builder = new StringBuilder();
            foreach (String k in keySet)
            {
                builder.Append(k + ",");
            }
            string seenLockIds = builder.ToString(0,builder.Length - 2);
            WriteLock(hlock, seenLockIds);
        }
        protected void WriteLock(HLock hlock, string seenLockIds)
        {
            Mutator<string> mutator = HFactory.CreateMutator(Keyspace, StringSerializer.Instance);
            mutator.AddInsertion(hlock.Path, LockManagerConfigurator.LockManagerCF,
            CreateColumnForLock(hlock.LockId, seenLockIds));
            mutator.Execute();
        }
        private HColumn<string, string> CreateColumnForLock(string name, string value)
        {
            return HFactory.CreateColumn(name, value, Keyspace.CreateClock(), colTtl, StringSerializer.Instance, StringSerializer.Instance);
        }
        private bool HasThisLockSeenMe(string commaSeparatedLockIds, string myLockId)
        {
            string[] seenLocksIds = commaSeparatedLockIds.Split(',');

            for (int i = 0; i < seenLocksIds.Length; i++)
            {
                if (seenLocksIds[i].Equals(myLockId))
                    return true;
            }
            return false;
        }
        protected void SetAcquired(HLock hlock, Dictionary<string, string> canBeEarlier)
        {
            Task t = StartTask(hlock);
        }
        protected async Task StartTask(HLock hlock)
        {
            await Task.Factory.StartNew(async () =>
            {
                Dictionary<string, string> existingLocks = new Dictionary<string, string>();
                string values = existingLocks[hlock.LockId];
                if (values == null)
                {
                }
                else
                {
                    WriteLock(hlock, values);
                }
                await Task.Delay(colTtl/2);
            }, CancellationToken.None, TaskCreationOptions.None, TaskScheduler.FromCurrentSynchronizationContext());
        }
        public override void Acquire(HLock hlock)
        {
            Acquire(hlock, long.MaxValue - (DateTime.Now.Ticks / TimeSpan.TicksPerMillisecond) - 10000);
        }
        public override void Acquire(HLock hlock, long timeout)
        {
            VerifyPrecondition(hlock);
            MaybeSetInternalLockId(hlock);
            WriteLock(hlock);

            Dictionary<string, string> canBeEarlier = ReadExistingLocks(hlock);
            
            string nextWaitingClientId = null;
            long waitStart = DateTime.Now.Ticks;
            
            while (true) 
            {
                if (canBeEarlier.Count <= 1)
                {
                    SetAcquired(hlock, canBeEarlier);
                    return;
                }
                
                if (waitStart + timeout < DateTime.Now.Ticks)
                {
                    DeleteLock(hlock);
                    throw new HLockTimeoutException(string.Format("Unable to get lock before {0} ", waitStart + timeout));
                }
                bool recv_all_acks = true;
                
                foreach (string otherLock in canBeEarlier.Keys)
                {
                    if (!hlock.LockId.Equals(otherLock) && !HasThisLockSeenMe(otherLock, hlock.LockId))
                    {
                        recv_all_acks = false;
                        break;
                    }
                }

                List<string> canBeEarlierSortedList = null;
                
                if (recv_all_acks)
                {
                    string[] keyArray = new List<string>(canBeEarlier.Keys).ToArray();
                    // sort them
                    Array.Sort<string>(keyArray);
                    nextWaitingClientId = canBeEarlierSortedList[0];
                    if (nextWaitingClientId.Equals(hlock.LockId))
                    {
                        break;
                    }
                }
                WriteLock(hlock, new List<string>(canBeEarlier.Keys));
                SmartWait(LockManagerConfigurator.BackOffRetryDelayInMillis);
                canBeEarlier = ReadExistingLocks(hlock);
            }
            
            SetAcquired(hlock, canBeEarlier);
        }
        public override void Release(HLock hlock)
        {
            VerifyPrecondition(hlock);
            DeleteLock(hlock);
            ((HLockImpl) hlock).IsAcquired = false;
        }
        private void VerifyPrecondition(HLock hlock)
        {
            if (hlock == null)
            {
            }
            if (hlock.Path == null)
            {
                //throw new RuntimeException("Lock path cannot be null");
            }
        }
        private void DeleteLock(HLock hlock)
        {
            // cancel the heartbeat task if it exists
            Task<string> heartbeat = ((HLockImpl) hlock).GetHeartBeat();
            if (heartbeat != null)
            {
                
            }
            Mutator<string> mutator = HFactory.CreateMutator(Keyspace, StringSerializer.Instance);
            mutator.AddDeletion(hlock.Path, LockManagerConfigurator.LockManagerCF, hlock.LockId,
                StringSerializer.Instance, Keyspace.CreateClock());
            mutator.Execute();
        }
        private void SmartWait(long sleepTime)
        {
            Random ran = new Random();
            long time = (sleepTime + (long)(ran.Next() * sleepTime));
            TimeSpan span = new TimeSpan(time);
            Thread.Sleep(span);
        }
	}
}